package com.offcn.bigdata.window

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
/**
 * @Auther: BigData-LGW
 * @ClassName: SlidingTimeWindow
 * @Date: 2020/12/17 17:05
 * @功能描述: $FunctionDescription
 * @Version:1.0
 */
/**
 * window之TimeWindow
 *     每隔多长时间进行一次计算，这个才是应用最广泛的
 *  分类：
 *      Tumbling TimeWindow :  滚动的时间窗口
 *              timeWindow(x)
 *              窗口时间固定，两个窗口之间没有缝隙，这种方式相当于len=sliding
 *      Sliding TimeWindow  :  滑动的时间窗口
 *              timeWindow(len, sliding) 相当于sparkstreaming中的Window操作
 *              每隔sliding对len时间内的数据进行统计分析
 *      Session TimeWindow  :  会话时间窗口
 */
object SlidingTimeWindow {
    def main(args: Array[String]): Unit = {
       val env = StreamExecutionEnvironment.getExecutionEnvironment
        val lines = env.socketTextStream("node-1",9999)
        val keyedStream = lines.flatMap(_.split("\\s+"))
            .map((_,1))
            .keyBy(_._1)
        keyedStream.timeWindow(Time.seconds(4),Time.seconds(2))
            .sum(1)
            .print()
        env.execute(s"${SlidingTimeWindow.getClass.getSimpleName}")
    }
}
